package com.navinfo.tripanalysis.service.impl;

import com.navinfo.tripanalysis.common.arithmetic.common.VehicleInfo;
import com.navinfo.tripanalysis.service.AbstractLoadVehicleInfoService;
import org.apache.commons.lang3.StringUtils;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.PairFlatMapFunction;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import scala.Tuple2;

import java.io.Serializable;
import java.util.*;
import java.util.stream.DoubleStream;

/**
 * 加载车辆的配置信息
 * @author 沈东生
 */
public class LoadVehicleInfoServiceImpl extends AbstractLoadVehicleInfoService {

    @Override
    protected String getTable() {
        return "(\tSELECT \n" +
                "    v.terminal_id terminal_id,\n" +
                "    ra.speed_ratio rearaxle_ratio,\n" +
                "    gb.speed_ratios gear_ratios,\n" +
                "    wl.radius wheel_radius,\n" +
                "    ng.eco_rotation eco_rotation,\n" +
                "    ng.max_torque  max_torque,\n" +
                "    weight.no wloadno,\n" +
                "    weight.empty_load,\n" +
                "    weight.half_load,\n" +
                "    weight.full_load\n" +
                "FROM tr_vehicle v\n" +
                "   LEFT JOIN tr_vehicle_gearbox gb ON v.gearbox_no = gb.no\n" +
                "   LEFT JOIN tr_vehicle_rearaxle ra ON v.rearaxle_no = ra.no\n" +
                "   LEFT JOIN tr_vehicle_wheel wl ON v.wheel_no = wl.no\n" +
                "   LEFT JOIN tr_vehicle_engine ng ON v.engine_no = ng.no\n" +
                "   LEFT JOIN tr_vehicle_weightload weight ON v.weightload_no = weight.no) t";
    }

    @Override
    protected JavaPairRDD<Long, VehicleInfo> getVehicleInfoPair(SparkSession session, JavaRDD<Row> rowJavaRDD) {
        logger.info("加载青汽车辆配置信息...");
        return rowJavaRDD.mapPartitionsToPair((PairFlatMapFunction<Iterator<Row>, Long, VehicleInfo>) LoadVehicleInfoServiceImpl::getVehicleInfoPair);
    }

    private static Iterator<Tuple2<Long, VehicleInfo>> getVehicleInfoPair(Iterator<Row> iterator) {
        List<Tuple2<Long, VehicleInfo>> vehicleInfos = new ArrayList<>();
        while (iterator.hasNext()) {
            Row row = iterator.next();
            VehicleInfo vehicleInfo = new VehicleInfo();

            long tid = row.getLong(0);
            String rearAxleRatioStr = row.getString(1);
            double rearAxleRatio = StringUtils.isEmpty(rearAxleRatioStr) ? 0 : Double.valueOf(rearAxleRatioStr);

            String gearRatios = row.getString(2);
            if (StringUtils.isNotEmpty(gearRatios)) {
                String[] split = gearRatios.split(",");
                DoubleStream doubleStream = Arrays.stream(split).mapToDouble(Double::valueOf);
                vehicleInfo.setGearsSpeedRatio(doubleStream.toArray());
            }

            String wheelRadiusStr = row.getString(3);
            double wheelRadius = StringUtils.isEmpty(wheelRadiusStr) ? 0 : Double.valueOf(wheelRadiusStr);
            String ecoRotationStr = row.getString(4);
            double ecoRotationLow = 0;
            double ecoRotationHigh = 0;
            if (StringUtils.contains(ecoRotationStr, "-")) {
                String[] ss = ecoRotationStr.split("-");
                ecoRotationLow = Double.valueOf(ss[0]);
                ecoRotationHigh = Double.valueOf(ss[1]);
            }
            String maxTorqueStr = row.getString(5);
            double maxTorque = StringUtils.isEmpty(maxTorqueStr) ? 0 : Double.valueOf(maxTorqueStr);
            String wloadNo=row.getString(6);


            vehicleInfo.setTerminalId(tid);
            vehicleInfo.setRearAxleSpeedRatio(rearAxleRatio);
            vehicleInfo.setWheelRadius(wheelRadius);
            vehicleInfo.setEngineMaxTorque(maxTorque);
            vehicleInfo.setEngineECORotationLow(ecoRotationLow);
            vehicleInfo.setEngineECORotationHigh(ecoRotationHigh);
            vehicleInfo.setWloadNo(wloadNo);
            vehicleInfo.setWloadEmpty(getInteger(row,"empty_load",null));
            vehicleInfo.setWloadHalf(getInteger(row,"half_load",null));
            vehicleInfo.setWloadFull(getInteger(row,"full_load",null));
            vehicleInfos.add(new Tuple2<>(tid, vehicleInfo));
        }
        return vehicleInfos.iterator();
    }

    private static Integer getInteger(Row rs,String colName,Integer defaultValue)  {
        Object obj = rs.getAs(colName);
        if(null != obj){
            try {
                return Integer.valueOf(obj.toString());
            }catch (Exception ex){
                ex.printStackTrace();
                return defaultValue;
            }
        }
        return defaultValue;
    }
}
